package com.gengzp.mq.server.manager.impl;

import com.gengzp.mq.server.facade.MqMessageFacadeService;
import com.gengzp.mq.server.manager.MqAckMessageManager;
import com.gengzp.mq.server.manager.MqResendMessageManager;
import com.gengzp.mq.server.model.message.entity.MqMessage;
import com.gengzp.mq.server.model.thread.MqThreadWaiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;

import static com.gengzp.mq.server.constants.MqRetryCountConstants.MAX_RETRY_COUNT;
import static com.gengzp.mq.server.constants.MqThreadNameConstants.MQ_MESSAGE_TIMEOUT_INSPECT_THREAD_NAME;
import static com.gengzp.mq.server.constants.MqWaitTimeConstants.WAIT_TIME;

@Component
public class MqAckMessageManagerImpl implements MqAckMessageManager {

    private static final Logger logger = LoggerFactory.getLogger(MqAckMessageManagerImpl.class);

    @Autowired
    private MqMessageFacadeService mqMessageFacadeService;

    @Autowired
    private MqResendMessageManager mqResendMessageManager;

    /**
     * 消费中的消息队列
     */
    private static final Queue<MqMessage> CONSUMING_MESSAGE_QUEUE = new ConcurrentLinkedQueue<>();

    /**
     * 线程等待对象
     */
    private static final MqThreadWaiter CHECK_TIMEOUT_WAITER = new MqThreadWaiter();

    @Override
    public void createMessageAckTimeoutInspectJob() {
        Thread existThread = CHECK_TIMEOUT_WAITER.getThread();
        if (existThread != null && existThread.isAlive()) {
            logger.error("消息ack超时检查线程已存在，无需重复创建");
            return;
        }

        // 线程初始化
        CHECK_TIMEOUT_WAITER.setThread(new Thread(() -> {
            while (true) {
                try {
                    // 判断队列是否为空, 队空时无限等待
                    synchronized (CHECK_TIMEOUT_WAITER) {
                        while (CONSUMING_MESSAGE_QUEUE.isEmpty()) {
                            CHECK_TIMEOUT_WAITER.wait(Long.MAX_VALUE);
                        }
                    }

                    // 当前系统时间
                    long currentTime = System.currentTimeMillis();
                    // 需要重发的消息集合
                    List<MqMessage> needResend = new ArrayList<>();
                    // 死信消息id集合
                    List<String> deathIds = new ArrayList<>();

                    // 查询出队列中已消费的消息id
                    Set<String> consumedIds = mqMessageFacadeService.searchConsumedSuccessMqMessagesByIds(
                                    CONSUMING_MESSAGE_QUEUE.stream().map(MqMessage::getMessageId).collect(Collectors.toList()))
                            .stream().map(MqMessage::getMessageId).collect(Collectors.toSet());

                    // 遍历处理消息
                    Iterator<MqMessage> iterator = CONSUMING_MESSAGE_QUEUE.iterator();
                    while (iterator.hasNext()) {
                        MqMessage msg = iterator.next();
                        // 移除已消费消息
                        if (consumedIds.contains(msg.getMessageId())) {
                            iterator.remove();
                            continue;
                        }

                        // 处理超时消息
                        if (currentTime - msg.getSendTimeStamp() >= WAIT_TIME) {
                            iterator.remove();
                            if (msg.getRetryCount() < MAX_RETRY_COUNT) {
                                needResend.add(msg);
                                logger.info("消息id: {} 超时未确认，即将重发; 当前重发次数: {}, 下次重发间隔时间: {}ms",
                                        msg.getMessageId(), msg.getRetryCount(), msg.getNextResendIntervalTime());
                            } else {
                                deathIds.add(msg.getMessageId());
                                logger.info("消息id: {} 超时未确认且达最大重试次数，即将加入死信队列", msg.getMessageId());
                            }
                        }
                    }

                    // 处理重发和死信
                    String resendIds = needResend.stream().map(MqMessage::getMessageId).collect(Collectors.joining(","));
                    if (!needResend.isEmpty()) { //重发
                        try {
                            mqResendMessageManager.resendMessages(needResend);
                        } catch (Exception e) {
                            logger.error("添加消息到重发队列失败，消息id: {}", resendIds, e);
                        }
                    }
                    if (!deathIds.isEmpty()) {  //死信
                        try {
                            mqMessageFacadeService.batchEditMessageToDeath(deathIds);
                            logger.info("死信消息处理成功，消息id: {}", String.join(",", deathIds));
                        } catch (Exception e) {
                            logger.error("死信消息处理失败，消息id: {}", String.join(",", deathIds), e);
                        }
                    }

                    // 等待时间计算
                    synchronized (CHECK_TIMEOUT_WAITER) {
                        if (!CONSUMING_MESSAGE_QUEUE.isEmpty()) {
                            long nextCheckTime = CONSUMING_MESSAGE_QUEUE.peek().getSendTimeStamp() + WAIT_TIME - currentTime;
                            CHECK_TIMEOUT_WAITER.wait(Math.max(nextCheckTime, 0));
                        }
                    }
                } catch (Exception e) {
                    logger.error("消息ack超时检查线程异常，将继续执行", e);
                }
            }
        }, MQ_MESSAGE_TIMEOUT_INSPECT_THREAD_NAME));

        // 启动线程
        CHECK_TIMEOUT_WAITER.getThread().start();
        logger.info("消息ack超时检查线程 {} 启动成功", MQ_MESSAGE_TIMEOUT_INSPECT_THREAD_NAME);
    }

    @Override
    public void sendMessageToAckTimeoutInspectJobQueue(MqMessage mqMessage) {
        CONSUMING_MESSAGE_QUEUE.add(mqMessage);
        synchronized (CHECK_TIMEOUT_WAITER) {
            CHECK_TIMEOUT_WAITER.notify();
        }
    }

    @Override
    public void sendMessageToAckTimeoutInspectJobQueue(List<MqMessage> mqMessages) {
        CONSUMING_MESSAGE_QUEUE.addAll(mqMessages);
        synchronized (CHECK_TIMEOUT_WAITER) {
            CHECK_TIMEOUT_WAITER.notify();
        }
        for (MqMessage mqMessage : mqMessages) {
            logger.info("消息id: {} 加入ack超时检查队列", mqMessage.getMessageId());
        }
    }

}